Close the streaming HTTP body on the Anthropic and OpenAI-Responses streaming paths - #718
Conversation
9b1f145 to
e188434
Compare
There was a problem hiding this comment.
Pull request overview
This pull request fixes a streaming resource-leak issue by ensuring SSE streams are explicitly closed in the Anthropic Messages and OpenAI Responses streaming implementations, aligning these paths with the existing OpenAI Chat Completions behavior and preventing HTTP connection leaks on early consumer termination or context cancellation.
Changes:
- Added
defer ...Close()immediately after creating streaming SSE streams inanthropicprovider/agent.goand both streaming branches ofopenaiprovider/responses.go. - Added black-box tests in each provider package to verify the underlying HTTP response body is closed when the consumer stops iteration early.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated no comments.
| File | Description |
|---|---|
| provider/openaiprovider/responses.go | Ensures OpenAI Responses streaming streams are closed on all exit paths (including early consumer exit). |
| provider/openaiprovider/responses_test.go | Adds a transport/body wrapper test to assert the streaming response body is closed when iteration ends early. |
| provider/anthropicprovider/agent.go | Ensures Anthropic Messages streaming stream is closed via defer to avoid leaking the HTTP body. |
| provider/anthropicprovider/agent_test.go | Adds a transport/body wrapper test to assert the streaming response body is closed when iteration ends early. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
This comment has been minimized.
This comment has been minimized.
github-actions
Bot
added
the
parity-approved
…aths The Anthropic Messages and OpenAI Responses streaming paths iterated the SSE stream but never called Close(), so on early consumer termination (yield returns false) or context cancellation the HTTP response body was never returned to the connection pool, leaking the underlying connection. Add a deferred stream.Close() immediately after each NewStreaming and GetStreaming call, mirroring the Chat Completions path which already does this and matching the .NET/Python SDKs that dispose the streaming response on early enumeration.
e188434 to
dd687df
Compare
Parity Review: No Issues FoundThis PR fixes a streaming HTTP body leak by adding Scope: Internal implementation fix only — no exported Go APIs were added, removed, or changed. Cross-repo parity: This is a convergence fix that matches .NET and Python SDK behavior (both dispose/close streaming responses on early enumeration exit). No parity issues. Labels: Warning Firewall blocked 1 domainThe following domain was blocked by the firewall during workflow execution:
network:
allowed:
- defaults
- "awmgmcpg"See Network Configuration for more information.
|
What
The Anthropic Messages streaming path (
anthropicprovider/agent.go) and the OpenAI Responses streaming paths (openaiprovider/responses.go) create an SSE stream viaNewStreaming/GetStreamingand iterate it withfor stream.Next(), but never callClose()on the stream. On early consumer termination (yieldreturnsfalsewhen the caller breaks out of the range) or context cancellation, the closure returns without releasing the HTTP response body.The SDK's
Stream.Close()releases the body (Stream.Close -> decoder.Close -> ReadCloser.Close). Without it, the body is never returned to the connection pool and the underlying connection leaks.This adds
defer func() { _ = stream.Close() }()immediately after each streaming call:anthropicprovider/agent.go: afterMessages.NewStreamingopenaiprovider/responses.go: afterResponses.GetStreaming(continuation-token resume) and afterResponses.NewStreamingWhy
The sibling Chat Completions path (
openaiprovider/chat.go) already does exactly this —stream := ...NewStreaming(...)immediately followed bydefer func() { _ = stream.Close() }(). This change brings the Anthropic and Responses paths in line with it. It also matches the .NET and Python SDKs, which dispose the streaming response when enumeration ends early, so streaming lifecycle semantics are consistent across the SDK ports.Testing
Added a black-box test to each package's canonical test file (
agent_test.go,responses_test.go) that:httptest.Server,http.ClientwhoseRoundTripperwraps the response body in aClose-countingReadCloser,Runand breaks out of the range after the first update,Close()was invoked.Each test fails before the fix (
Closecount 0) and passes after (count 1).go build ./...,go vet, andgo testpass for both changed packages.